package org.spiderflow.core.executor.shape;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.spiderflow.common.enums.PauseEvent;
import org.spiderflow.common.executor.Grammerable;
import org.spiderflow.common.context.SpiderContext;
import org.spiderflow.core.exception.SpiderException;
import org.spiderflow.core.utils.DataSourceUtils;
import org.spiderflow.core.utils.ExpressionUtils;
import org.spiderflow.core.utils.ExtractUtils;
import org.spiderflow.common.executor.ShapeExecutor;
import org.spiderflow.common.model.Grammar;
import org.spiderflow.common.model.SpiderNode;
import org.springframework.jdbc.core.ArgumentPreparedStatementSetter;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.support.GeneratedKeyHolder;
import org.springframework.jdbc.support.KeyHolder;
import org.springframework.stereotype.Component;

import java.lang.reflect.Array;
import java.sql.PreparedStatement;
import java.sql.Statement;
import java.util.*;

/**
 * SQL执行器
 *
 * @author jmxd
 */
@Component
public class ExecuteSqlExecutor implements ShapeExecutor, Grammerable {

    private static final String DATASOURCE_ID = "datasourceId";

    private static final String SQL = "sql";

    private static final String STATEMENT_TYPE = "statementType";

    private static final String STATEMENT_SELECT = "select";

    private static final String STATEMENT_SELECT_ONE = "selectOne";

    private static final String STATEMENT_SELECT_INT = "selectInt";

    private static final String STATEMENT_INSERT = "insert";

    private static final String STATEMENT_UPDATE = "update";

    private static final String STATEMENT_DELETE = "delete";
    private static final String SELECT_RESULT_STREAM = "isStream";
    private static final String STATEMENT_INSERT_PK = "insertofPk";

    private static final Logger logger = LoggerFactory.getLogger(ExecuteSqlExecutor.class);

    @Override
    public void execute(SpiderNode node, SpiderContext context, Map<String, Object> variables) {
        String dsId = node.getStringJsonValue(DATASOURCE_ID);
        String sql = node.getStringJsonValue(SQL);
        if (!checkParam(dsId, sql)) {
            return;
        }
        JdbcTemplate template = new JdbcTemplate(DataSourceUtils.getDataSource(dsId));
        //根据占位符提取变量名
        List<String> parameters = ExtractUtils.getMatchers(sql, "#(.*?)#", true);
        String finalSql = buildSql(sql, node.getNodeId(), context, variables);
        if (StringUtils.isBlank(finalSql)) {
            return;
        }
        int size = parameters.size();
        Object[] params = new Object[size];
        boolean hasList = false;
        int parameterSize = 0;
        //当参数中存在List或者数组时，认为是批量操作
        for (int i = 0; i < size; i++) {
            Object parameter = ExpressionUtils.execute(parameters.get(i), variables);
            if (parameter != null) {
                if (parameter instanceof List) {
                    hasList = true;
                    parameterSize = Math.max(parameterSize, ((List<?>) parameter).size());
                } else if (parameter.getClass().isArray()) {
                    hasList = true;
                    parameterSize = Math.max(parameterSize, Array.getLength(parameter));
                }
            }
            params[i] = parameter;
        }
        Object rs = executeSql(node, finalSql, template, params, hasList, parameterSize);
        variables.put("rs", rs);
    }

    private static boolean checkParam(String dsId, String sql) {
        if (StringUtils.isBlank(dsId)) {
            logger.warn("数据源ID为空！");
            return false;
        } else if (StringUtils.isBlank(sql)) {
            logger.warn("sql为空！");
            return false;
        } else {
            return true;
        }
    }

    private static String buildSql(String originalSql, String nodeId, SpiderContext context, Map<String, Object> variables) {
        //把变量替换成占位符
        String sql = ExtractUtils.replaceAll(originalSql, "#(.*?)#", "?");
        try {
            Object sqlObject = ExpressionUtils.execute(sql, variables);
            if (sqlObject == null) {
                logger.warn("获取的sql为空！");
                return "";
            }
            String finalSql = sqlObject.toString();
            context.pause(nodeId, PauseEvent.COMMON, SQL, finalSql);
            return finalSql;
        } catch (RuntimeException e) {
            logger.error("获取sql出错,异常信息:{}", e.getMessage(), e);
            ExceptionUtils.wrapAndThrow(e);
            return "";
        }
    }

    private static Object executeSql(SpiderNode node, String sql, JdbcTemplate template, Object[] params,
                                     boolean hasList, int parameterSize) {
        String statementType = node.getStringJsonValue(STATEMENT_TYPE);
        logger.debug("执行sql：{}", sql);
        Object rs;
        try {
            if (STATEMENT_SELECT.equals(statementType)) {
                rs = executeSelect(node, sql, template, params);
            } else if (STATEMENT_SELECT_ONE.equals(statementType)) {
                rs = executeSelectOne(sql, template, params);
            } else if (STATEMENT_SELECT_INT.equals(statementType)) {
                rs = executeSelectInt(sql, template, params);
            } else if (STATEMENT_UPDATE.equals(statementType) || STATEMENT_INSERT.equals(statementType)
                    || STATEMENT_DELETE.equals(statementType)) {
                rs = executeUpdate(sql, template, params, hasList, parameterSize);
            } else if (STATEMENT_INSERT_PK.equals(statementType)) {
                rs = executeInsertAndReturnPk(sql, template, params);
            } else {
                throw new SpiderException("没有匹配的操作类型：" + statementType);
            }
            return rs;
        } catch (RuntimeException e) {
            logger.error("执行sql出错,异常信息:{}", e.getMessage(), e);
            ExceptionUtils.wrapAndThrow(e);
            return null;
        }
    }

    private static Integer executeUpdate(String sql, JdbcTemplate template, Object[] params, boolean hasList, int parameterSize) {
        int updateCount = 0;
        if (hasList) {
            // 批量操作时，将参数Object[]转化为List<Object[]>
            // 当参数不为数组或List时，自动转为Object[]
            // 当数组或List长度不足时，自动取最后一项补齐
            int[] rs = template.batchUpdate(sql, convertParameters(params, parameterSize));
            if (rs.length > 0) {
                updateCount = Arrays.stream(rs).sum();
            }
        } else {
            updateCount = template.update(sql, params);
        }
        return updateCount;
    }

    private static Integer executeSelectInt(String sql, JdbcTemplate template, Object[] params) {
        return template.queryForObject(sql, params, Integer.class);
    }

    private static Map<String, Object> executeSelectOne(String sql, JdbcTemplate template, Object[] params) {
        return template.queryForMap(sql, params);
    }

    private static Object executeSelect(SpiderNode node, String sql,
                                        JdbcTemplate template, Object[] params) {
        boolean isStream = "1".equals(node.getStringJsonValue(SELECT_RESULT_STREAM));
        if (isStream) {
            return template.queryForRowSet(sql, params);
        } else {
            return template.queryForList(sql, params);
        }
    }

    private static Integer executeInsertAndReturnPk(String sql, JdbcTemplate template, Object[] params) {
        KeyHolder keyHolder = new GeneratedKeyHolder();
        template.update(con -> {
            PreparedStatement ps = con.prepareStatement(sql, Statement.RETURN_GENERATED_KEYS);
            new ArgumentPreparedStatementSetter(params).setValues(ps);
            return ps;
        }, keyHolder);
        Number holderKey = keyHolder.getKey();
        return holderKey == null ? -1 : holderKey.intValue();
    }

    private static List<Object[]> convertParameters(Object[] params, int length) {
        List<Object[]> result = new ArrayList<>(length);
        int size = params.length;
        for (int i = 0; i < length; i++) {
            Object[] parameters = new Object[size];
            for (int j = 0; j < size; j++) {
                parameters[j] = getValue(params[j], i);
            }
            result.add(parameters);
        }
        return result;
    }

    private static Object getValue(Object object, int index) {
        if (object == null) {
            return null;
        } else if (object instanceof List) {
            List<?> list = (List<?>) object;
            int size = list.size();
            if (size > 0) {
                return list.get(Math.min(list.size() - 1, index));
            }
        } else if (object.getClass().isArray()) {
            int size = Array.getLength(object);
            if (size > 0) {
                Array.get(object, Math.min(-1, index));
            }
        } else {
            return object;
        }
        return null;
    }

    @Override
    public String supportShape() {
        return "executeSql";
    }

    @Override
    public List<Grammar> grammars() {
        Grammar grammar = new Grammar();
        grammar.setComment("执行SQL结果");
        grammar.setFunction("rs");
        grammar.setReturns(Arrays.asList("List<Map<String,Object>>", "int"));
        return Collections.singletonList(grammar);
    }
}
